package com.mini.gateway.core.config.exception;

import com.alibaba.fastjson.JSON;
import com.mini.gateway.core.modules.common.KafkaService;
import com.mini.gateway.core.modules.log.entity.ApiLog;
import com.mini.gateway.core.modules.log.service.ApiLogConstants;
import org.apache.commons.lang.StringUtils;
import org.springframework.boot.autoconfigure.web.ErrorProperties;
import org.springframework.boot.autoconfigure.web.ResourceProperties;
import org.springframework.boot.autoconfigure.web.reactive.error.DefaultErrorWebExceptionHandler;
import org.springframework.boot.web.reactive.error.ErrorAttributes;
import org.springframework.context.ApplicationContext;
import org.springframework.http.HttpStatus;
import org.springframework.http.InvalidMediaTypeException;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.server.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.LocalDateTime;
import java.util.*;

import static org.springframework.web.reactive.function.server.RequestPredicates.all;
import static org.springframework.web.reactive.function.server.RouterFunctions.route;

/**
 * JsonExceptionHandler
 *
 */
public class JsonExceptionHandler extends DefaultErrorWebExceptionHandler {

    private static final Map<HttpStatus.Series, String> SERIES_VIEWS;

    static {
        Map<HttpStatus.Series, String> views = new EnumMap<>(HttpStatus.Series.class);
        views.put(HttpStatus.Series.CLIENT_ERROR, "4xx");
        views.put(HttpStatus.Series.SERVER_ERROR, "5xx");
        SERIES_VIEWS = Collections.unmodifiableMap(views);
    }

    private final ErrorProperties errorProperties;
    private final KafkaService kafkaService;

    /**
     * Create a new {@code DefaultErrorWebExceptionHandler} instance.
     * @param errorAttributes the error attributes
     * @param resourceProperties the resources configuration properties
     * @param errorProperties the error configuration properties
     * @param applicationContext the current application context
     */
    public JsonExceptionHandler(ErrorAttributes errorAttributes,
                                           ResourceProperties resourceProperties, ErrorProperties errorProperties,
                                           ApplicationContext applicationContext, KafkaService kafkaService) {
        super(errorAttributes, resourceProperties, errorProperties, applicationContext);
        this.errorProperties = errorProperties;
        this.kafkaService = kafkaService;
    }

    @Override
    protected RouterFunction<ServerResponse> getRoutingFunction(
            ErrorAttributes errorAttributes) {
        return route(acceptsTextHtml(), this::renderErrorView).andRoute(all(),
                this::renderErrorResponse);
    }

    /**
     * Render the error information as an HTML view.
     * @param request the current request
     * @return a {@code Publisher} of the HTTP response
     */
    protected Mono<ServerResponse> renderErrorView(ServerRequest request) {
        boolean includeStackTrace = isIncludeStackTrace(request, MediaType.TEXT_HTML);
        Map<String, Object> error = getErrorAttributes(request, includeStackTrace);
        HttpStatus errorStatus = getHttpStatus(error);
        ServerResponse.BodyBuilder responseBody = ServerResponse.status(errorStatus)
                .contentType(MediaType.TEXT_HTML);
        return Flux
                .just("error/" + errorStatus.value(),
                        "error/" + SERIES_VIEWS.get(errorStatus.series()), "error/error")
                .flatMap((viewName) -> renderErrorView(viewName, responseBody, error))
                .switchIfEmpty(this.errorProperties.getWhitelabel().isEnabled()
                        ? renderDefaultErrorView(responseBody, error)
                        : Mono.error(getError(request)))
                .next();
    }

    /**
     * Render the error information as a JSON payload.
     * @param request the current request
     * @return a {@code Publisher} of the HTTP response
     */
    protected Mono<ServerResponse> renderErrorResponse(ServerRequest request) {
        boolean includeStackTrace = isIncludeStackTrace(request, MediaType.ALL);
        Map<String, Object> error = getErrorAttributes(request, includeStackTrace);
        HttpStatus httpStatus = getHttpStatus(error);
        String serialNo = request.exchange().getAttribute(ApiLogConstants.API_LOG_SERIAL);
        if (StringUtils.isNotEmpty(serialNo)) {
            ApiLog apiLog = new ApiLog();
            apiLog.setSerialNo(serialNo);
            apiLog.setResponseCode(httpStatus.value());
            apiLog.setEndTime(LocalDateTime.now());
            String json = JSON.toJSONString(apiLog);
            // 日志投递到kafka，做记录、预警处理
            kafkaService.sendMessage(ApiLogConstants.KAFKA_TOPIC_APILOG, serialNo, json);
        }
        return ServerResponse.status(getHttpStatus(error))
                .contentType(MediaType.APPLICATION_JSON_UTF8)
                .body(BodyInserters.fromObject(error));
    }

    /**
     * Determine if the stacktrace attribute should be included.
     * @param request the source request
     * @param produces the media type produced (or {@code MediaType.ALL})
     * @return if the stacktrace attribute should be included
     */
    protected boolean isIncludeStackTrace(ServerRequest request, MediaType produces) {
        ErrorProperties.IncludeStacktrace include = this.errorProperties
                .getIncludeStacktrace();
        if (include == ErrorProperties.IncludeStacktrace.ALWAYS) {
            return true;
        }
        if (include == ErrorProperties.IncludeStacktrace.ON_TRACE_PARAM) {
            return isTraceEnabled(request);
        }
        return false;
    }

    /**
     * Get the HTTP error status information from the error map.
     * @param errorAttributes the current error information
     * @return the error HTTP status
     */
    protected HttpStatus getHttpStatus(Map<String, Object> errorAttributes) {
        int statusCode = (int) errorAttributes.get("status");
        return HttpStatus.valueOf(statusCode);
    }

    /**
     * Predicate that checks whether the current request explicitly support
     * {@code "text/html"} media type.
     * <p>
     * The "match-all" media type is not considered here.
     * @return the request predicate
     */
    protected RequestPredicate acceptsTextHtml() {
        return (serverRequest) -> {
            try {
                List<MediaType> acceptedMediaTypes = serverRequest.headers().accept();
                acceptedMediaTypes.remove(MediaType.ALL);
                MediaType.sortBySpecificityAndQuality(acceptedMediaTypes);
                return acceptedMediaTypes.stream()
                        .anyMatch(MediaType.TEXT_HTML::isCompatibleWith);
            }
            catch (InvalidMediaTypeException ex) {
                return false;
            }
        };
    }

}
